package com.intct.hbase.dml;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.regionserver.RegionSplitPolicy;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * @author gufg
 * @since 2025-08-06 10:49
 *
 * HBase中内置过滤类：
 *        DependentColumnFilter
 *        KeyOnlyFilter
 *        ColumnCountGetFilter
 *        SingleColumnValueFilter           单值过滤
 *        PrefixFilter                      前缀过滤
 *        SingleColumnValueExcludeFilter
 *        FirstKeyOnlyFilter
 *        ColumnRangeFilter
 *        ColumnValueFilter
 *        TimestampsFilter
 *        FamilyFilter
 *        QualifierFilter                   列名过滤
 *        ColumnPrefixFilter
 *        RowFilter                         rowkey过滤
 *        MultipleColumnPrefixFilter
 *        InclusiveStopFilter
 *        PageFilter                        分页过滤
 *        ValueFilter                       值过滤
 *        ColumnPaginationFilter
 */
public class ScanCommand {
    public static void main(String[] args) throws IOException {
        // 创建HBase连接
        Connection connection = ConnectionFactory.createConnection();

        // 设置TableName
        TableName tableName = TableName.valueOf("intct83", "t5");

        // 获取Admin实例
        Admin admin = connection.getAdmin();
        Table table = null;

        try {
            // 判断表存在
            if (admin.tableExists(tableName)) {
                // 获取Table实例
                table = connection.getTable(tableName);

                // Scan对象
                Scan scan = new Scan();

//                // 开始rowkey和结束rowkey
//                scan.withStartRow("1001".getBytes());
//                scan.withStopRow("1003".getBytes());
//
//                // 检索版本
//                scan.readAllVersions();

                // 过滤条件
                FilterList filterList = new FilterList();

                // 单值过滤
                Filter singleColumnValueFilter = new SingleColumnValueFilter(
                        // 列族
                        "f1".getBytes(),
                        // 列限定符
                        "age".getBytes(),
                        // 比较符
                        CompareOperator.GREATER_OR_EQUAL,
                        // 值
                        "19".getBytes()
                );

                // rowkey过滤
                RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, new RegexStringComparator("1001.*"));

                // 增加多个过滤器
                filterList.addFilter(singleColumnValueFilter);
                filterList.addFilter(rowFilter);

                scan.setFilter(filterList);

                // 检索数据
                ResultScanner scanner = table.getScanner(scan);

                for (Result result : scanner) {
                    Cell[] cells = result.rawCells();
                    for (Cell cell : cells) {
                        // Cell == rowkey +  列族 + 列限定符 + 值
                        String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                        String family = Bytes.toString(CellUtil.cloneFamily(cell));
                        String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                        String value = Bytes.toString(CellUtil.cloneValue(cell));
                        System.out.println("rowkey = " + rowkey + "; family = " + family + "; qualifier = " + qualifier + "; value = " + value);
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭资源
            if (table != null) {
                table.close();
            }
            if (admin != null) {
                admin.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
    }
}
